package com.exm.collectcodenew.kafka.producer;

import com.exm.collectcodenew.kafka.producer.customSerializer.User;
import com.exm.collectcodenew.kafka.producer.customSerializer2.SQLData;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class ProducerTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "59.110.35.177:9092");//必须指定
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//必须指定
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//必须指定
        props.put("acks", "-1");
        props.put("retries", 3);
        props.put("batch.size", 323840);
        props.put("linger.ms", 10);
        props.put("buffer.memory", 33554432);
        props.put("max.block.ms", 3000);
        props.put("partitioner.class","com.exm.collectcodenew.kafka.producer.customPartitioner.AuditPartitioner");
        //构建拦截链
        List<String> interceptors =new ArrayList<>();
        interceptors.add("com.exm.collectcodenew.kafka.producer.customInterceptor.TimeStampPrependerInterceptor");
        interceptors.add("com.exm.collectcodenew.kafka.producer.customInterceptor.CounterInterceptor");
        props.put("interceptor.classes",interceptors);
        Producer<String, String> producer = new KafkaProducer<>(props);
//        //构建User对象
//        User user = User.builder()
//                .firstName("Z")
//                .lastName("tt")
//                .age(18)
//                .address("beijing")
//                .build();
        for (int i = 0; i < 10; i++) {
            ProducerRecord record = new ProducerRecord("topic-test1","test_"+i);
            producer.send(record).get();
        }


//        ProducerRecord nonKeyRecord = new ProducerRecord("topic-test","non-key record");
//        ProducerRecord auditRecord = new ProducerRecord("topic-test", "audit","audit record");
//        ProducerRecord nonAuditRecord =new ProducerRecord("topic-test","other","non-sudit record");
//        producer.send(nonKeyRecord).get();
//        producer.send(nonAuditRecord).get();
//        producer.send(auditRecord).get();
//        producer.send(nonKeyRecord).get();
//        producer.send(nonAuditRecord).get();
//        for (int i = 0; i < 100; i++) {
//          producer.send(new ProducerRecord<>("ztt", Integer.toString(i), Integer.toString(i)));
//        }
        producer.close();
    }
}
